package registry

import (
	"fmt"
	. "gitee.com/aldrice/monitor/configs"
	"gitee.com/aldrice/monitor/pkg/logging"
	"github.com/streadway/amqp"
	"go.uber.org/zap"
)

//生产者：接受模块  消费者：数据处理
var dataProcess *RabbitMQ

// 生产者：数据处理模块  消费者：邮件发送模块
var sendEmail *RabbitMQ

//rabbitMQ结构体
type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列名称
	QueueName string
	//交换机名称
	Exchange string
	//bind Key 名称
	Key string
	//连接信息
	Mqurl string
}

//创建简单模式的RabbitMQ实例
func NewRabbitMQSimple(queueName string) (*RabbitMQ, error) {
	//创建RabbitMQ实例
	rabbitmq := &RabbitMQ{QueueName: queueName, Exchange: "", Key: "", Mqurl: Config.RabbitMQ.URL}
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.Mqurl)
	if err != nil {
		logging.Logger().Error("failed to connect rabbitmq!", zap.Error(err))
		return nil, err
	}
	//获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	if err != nil {
		logging.Logger().Error("failed to open a channel", zap.Error(err))
		return nil, err
	}
	return rabbitmq, nil
}

// InitRabbitMQ 初始化结构体实例
func InitRabbitMQ() error {
	var err error
	dataProcess, err = NewRabbitMQSimple("dataProcess")
	if err != nil {
		return err
	}
	sendEmail, err = NewRabbitMQSimple("sendEmail")
	if err != nil {
		return err
	}
	return nil
}

func GetDataProcessMQ() *RabbitMQ {
	return dataProcess
}
func GetSendEmailMQ() *RabbitMQ {
	return sendEmail
}

//断开channel 和 connection
func (r *RabbitMQ) Destory() error {
	err := r.channel.Close()
	if err != nil {
		return err
	}
	return r.conn.Close()
}

// 生产者发布消息
func (r *RabbitMQ) PublishSimple(message []byte) error {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	_, err := r.channel.QueueDeclare(
		r.QueueName,
		//是否持久化
		true,
		//是否自动删除
		false,
		//是否具有排他性
		false,
		//是否阻塞处理
		false,
		//额外的属性
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	//调用channel 发送消息到队列中
	return r.channel.Publish(
		r.Exchange,
		r.QueueName,
		//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false,
		//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        message,
		})
}

type Module interface {
	Behavior(msg amqp.Delivery) (string, error)
}

// 消费者消费消息
func (r *RabbitMQ) ConsumeSimple(m Module) error {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	q, err := r.channel.QueueDeclare(
		r.QueueName,
		//是否持久化
		true,
		//是否自动删除
		false,
		//是否具有排他性
		false,
		//是否阻塞处理
		false,
		//额外的属性
		nil,
	)
	if err != nil {
		return err
	}

	//接收消息
	msgs, err := r.channel.Consume(
		q.Name, // queue
		//用来区分多个消费者
		"", // consumer
		//是否自动应答
		true, // auto-ack
		//是否独有
		false, // exclusive
		//设置为true，表示 不能将同一个Conenction中生产者发送的消息传递给这个Connection中 的消费者
		false, // no-local
		//列是否阻塞
		false, // no-wait
		nil,   // args
	)
	if err != nil {
		return err
	}

	//启用协程处理消息
	go func() {
		for d := range msgs {
			message, e := m.Behavior(d)
			if e != nil {
				logging.Logger().Error(message, zap.Error(e))
			}
		}
	}()
	return nil
}
